import * as thread from 'worker_threads';
import { EventEmitter } from 'events';
import { debuglog } from 'util';

const debug = debuglog('threadPool');

export interface Message {
  /** 消息类型 */
  type: string;
  [key: string]: any;
}
export interface FinishMessage extends Message {
  /** 任务完成耗时 */
  takeUpTime: number;
  error?: string;
}

interface MessageEvent {
  on(event: 'finish', listener: (worker: thread.Worker, message: FinishMessage) => void): void;
  on(event: 'cleanEnd', listener: (worker: thread.Worker, message: Message) => void): void;
  on(event: string, listener: (...args: any[]) => void): void;
  emit(event: string | symbol, ...args: any[]): boolean;
}
export interface Task {
  /** 任务唯一id */
  id: string;
  /** 任务文件的绝对路径 */
  executableFilePath: string;
  /** 执行参数 */
  args: any[];
  /** 重试次数 */
  retry?: number;
}

export class ThreadPool {
  /** 线程池开关 */
  public run = true;

  /** 线程消息事件 */
  public messageEvent: MessageEvent = new EventEmitter();

  /** 失败的任务 */
  public errorTaskList: Task[] = [];

  /** 任务默认重试次数，如果task中未指定retry，则为默认值 */
  public retryCount = 3;

  /** 空闲线程销毁的定时器 */
  private destroyTimer = new Map<thread.Worker, NodeJS.Timeout>();

  /** 空闲的线程 */
  private idlePool = new Set<thread.Worker>();

  /** 统计所有的工作线程 */
  private workerPool = new Set<thread.Worker>();

  /** 待处理任务队列 */
  private undealTaskList: Task[] = [];

  private agentMap = new Map<string, thread.Worker>();

  /**
   * 创建线程池
   * @param maxThread 线程池最大值
   * @param keepAliveTime 空闲线程存活时间(单位: 毫秒)，如果设置为0则表示不进行回收
   */
  constructor(public maxThread: number, private keepAliveTime = 10000) {
    this.messageEvent.on('finish', (worker, message) => {
      debug(`任务 ${ worker.task.id }完成耗时(${ message.takeUpTime })ms`);
      if (!this.run) {
        // 线程池有序退出
        debug(`${ worker.task.id } 手动退出！`);
        worker.terminate();
        return;
      }
      const task = this.getTask(worker.task.executableFilePath);
      if (task) {
        // 找到同工种任务，则直接运行
        worker.task = task;
        worker.postMessage({ type: 'run', args: task.args });
        return;
      }
      // 没有同工种任务则进入销毁期
      this.destoryStage(worker);
    });

    this.messageEvent.on('ready', (worker) => {
      worker.postMessage({ type: 'run', args: worker.task.args });
    });

    this.messageEvent.on('agent', (message) =>{
      const worker = this.agentMap.get(message.name);
      if (worker) {
        worker.postMessage({ type: 'receive', data: message.data });
      }
    });

    this.messageEvent.on('worker', (message) =>{
      const data = { type: 'receive', data: message.data };
      for (const worker of this.workerPool) {
        worker.postMessage(data);
      }
    });
  }

  public addTask(task: Task) {
    task.retry = typeof task.retry === 'undefined' ? this.retryCount : task.retry;
    if (this.idlePool.size > 0) {
      let defaultWorker: thread.Worker;
      for (const worker of this.idlePool) {
        if (worker.task.executableFilePath === task.executableFilePath) {
          worker.task = task;
          // 取消线程空闲标记
          this.idlePool.delete(worker);
          // 清除销毁定时器
          clearTimeout(this.destroyTimer.get(worker));
          this.destroyTimer.delete(worker);
          // 运行任务
          worker.postMessage({ type: 'run', args: task.args });
          return;
        }
        if (!worker.task.executableFilePath) {
          defaultWorker = worker;
        }
      }
      if (defaultWorker) {
        // 清除自动销毁的定时器
        clearTimeout(this.destroyTimer.get(defaultWorker));
        this.destroyTimer.delete(defaultWorker);
        // 取消线程空闲标记
        this.idlePool.delete(defaultWorker);
        // 标记线程工种
        defaultWorker.task = task;
        // 分配任务
        defaultWorker.postMessage({ type: 'before', task });
        return;
      }
    }
    if (this.workerPool.size < this.maxThread) {
      this.createWorker(task);
      return;
    }
    this.undealTaskList.push(task);
  }

  /**
   * 优雅的停止
   * 工作进程会在当前任务完成后进入空闲状态，不再接收新的任务
   */
  public stop() {
    this.run = false;
    for (const worker of this.idlePool) {
      worker.terminate();
    }
    this.idlePool.clear();
  }

  /**
   * 重新分配任务
   */
  public start() {
    this.run = true;
    while (this.undealTaskList.length !== 0) {
      if (this.workerPool.size === this.maxThread) {
        break;
      }
      this.createWorker(this.undealTaskList.pop());
    }
  }

  /**
   * 创建agent线程
   * @param name agent名称
   * @param executableFilePath 任务文件的绝对路径
   */
  public createAgent(name: string, executableFilePath: string) {
    if (this.agentMap.get(name)) {
      throw new Error(`${ name } exists！`);
    }
    const worker = new thread.Worker(executableFilePath);

    worker.on('message', (message: Message) => {
      this.messageEvent.emit(message.type, worker, message);
    });

    worker.on('online', () => {
      worker.postMessage({ type: 'run' });
    });

    worker.on('error', (error) => {
      debug(`agent线程 ${ name } 出现异常 ${ error }`);
    });

    worker.on('exit', () => {
      debug(`agent线程退出，即将重启！`);
      this.createAgent(name, executableFilePath);
    });
  }

  /**
   * 获取待处理任务数量
   */
  public getUndealTaskCount() {
    return this.undealTaskList.length;
  }

  /**
   * 根据工种查询对应的待处理任务，如果不存在则返回最近的任务
   * @param type 工种
   */
  private getTask(executableFilePath: string): Task | undefined {
    for (let i = 0; i < this.undealTaskList.length; i ++) {
      const task = this.undealTaskList[i];
      if (task.executableFilePath === executableFilePath) {
        this.undealTaskList.splice(i, 1);
        return task;
      }
    }
  }

  private createWorker(task: Task) {
    const worker = new thread.Worker(task.executableFilePath);
    worker.task = task;
    this.workerPool.add(worker);
    worker.on('message', (message: Message) => {
      this.messageEvent.emit(message.type, worker, message);
    });

    worker.on('online', () => {
      worker.postMessage({ type: 'before' });
    });

    worker.on('error', (error) => {
      const task = worker.task;
      debug(`${ task.id } 出现异常 ${ error }`);
      task.retry --;
      if (task.retry <= 0) {
        debug(`${ task.id }重试次数达到上限!`);
        this.errorTaskList.push(task);
        return;
      }
      debug(`${ task.id } 还有${ task.retry }次重试机会`);
      // 失败任务重回队列等待重试
      this.addTask(task);
    });

    worker.on('exit', () => {
      this.idlePool.delete(worker);
      this.workerPool.delete(worker);
      debug(`线程退出，当前剩余线程数 ${ this.workerPool.size }`);
      while (this.undealTaskList.length > 0 && this.workerPool.size < this.maxThread) {
        const task = this.undealTaskList.shift();
        this.createWorker(task);
      }
    });
  }

   /**
   * 工作线程进入销毁期
   * @param worker 工作线程
   */
  private destoryStage(worker: thread.Worker) {
    this.idlePool.add(worker);
    if (this.keepAliveTime) {
      // 创建销毁计时器，空闲线程
      this.destroyTimer.set(worker, setTimeout(() => {
        this.idlePool.delete(worker);
        this.destroyTimer.delete(worker);
        worker.terminate();
      }, this.keepAliveTime));
    }
  }
}

export class Agent {
  private debug = debuglog('agentThread');

  constructor () {
    thread.parentPort.on('message', async (message) => {
      switch (message.type) {
        case 'run':
          await this.run();
          break;
        case 'receive':
          this.receiveMessage(message.data);
          break;
        default:
          this.debug(`未知消息: ${ JSON.stringify(message) }`);
          break;
      }
    });
  }

  /**
   * 线程执行任务
   * @param _args 参数表
   */
  public async run() {
    throw new Error('run: interface not implemented');
  }

  /**
   * 接受通知
   * @param _data 
   */
  public receiveMessage(_data: Object) {}

  public sendMessage(message: Message) {
    thread.parentPort.postMessage(message);
  }

  public sendToAgent(name: string, data: Object) {
    thread.parentPort.postMessage({ type: 'agent', name, data });
  }

  public sendToWorker(data: Object) {
    thread.parentPort.postMessage({ type: 'worker', data });
  }
}

export class Worker {
  private debug = debuglog('workerThread');

  constructor () {
    thread.parentPort.on('message', async (message) => {
      switch (message.type) {
        case 'before':
          try {
            await this.before();
          } catch (error) {
            this.debug('before', error);  
          }
          thread.parentPort.postMessage({ type: 'ready' });
          break;
        case 'run':
          const finishMessage: FinishMessage = { type: 'finish', takeUpTime: 0 };
          const beginTime = Date.now();
          try {
            await this.run(...message.args);
          } catch (error) {
            finishMessage.error = error.toString();
            this.debug('任务异常', error);
          }
          finishMessage.takeUpTime = Date.now() - beginTime;
          thread.parentPort.postMessage(finishMessage);
          break;
        case 'clean':
          try {
            await this.clean();
          } catch (error) {
            this.debug('clean', error);
          }
          thread.parentPort.postMessage({ type: 'cleanEnd' });
          break;
        case 'receive':
          this.receiveMessage(message.data);
          break;
        default:
          this.debug(`未知消息: ${ JSON.stringify(message) }`);
          break;
      }
    });
  }

  /**
   * 线程准备工作
   */
  public async before() { }

  /**
   * 线程执行任务
   * @param _args 参数表
   */
  public async run(..._args) {
    throw new Error('run: interface not implemented');
  }

  /**
   * 清理线程环境
   */
  public async clean() { }

  /**
   * 接受通知
   * @param _data 
   */
  public receiveMessage(_data: Object) {}

  public sendMessage(message: Message) {
    thread.parentPort.postMessage(message);
  }

  public sendToAgent(name: string, data: Object) {
    thread.parentPort.postMessage({ type: 'agent', name, data });
  }

  public sendToWorker(data: Object) {
    thread.parentPort.postMessage({ type: 'worker', data });
  }
}